package KafkaTest

import (
	"bytes"
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"log"
)

var (
	KafkaProducer sarama.SyncProducer
)

func ProducerInit() {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForLocal
	config.Producer.Retry.Max = 5
	config.Producer.Return.Successes = true
	brokers := KafkaConfigs.KafkaBrokers
	var err error
	KafkaProducer, err = sarama.NewSyncProducer(brokers, config)
	if err != nil {
		fmt.Printf("Kafka生产者程序异常 -> %v \n", err)
		panic(err.Error())
	}
	log.Printf("Kafka生产者初始化成功 -> %v, topic -> %v, \n", KafkaProducer, KafkaConfigs.KafkaTopic)
}

//解析为json字符串
func dumpString(v interface{}) (str string) {

	bs, err := json.Marshal(v)
	b := bytes.Buffer{}
	if err != nil {
		b.WriteString("{err:\"json format error.")
		b.WriteString(err.Error())
		b.WriteString("\"}")
	} else {
		b.Write(bs)
	}
	str = b.String()
	return str
}
